package a_aa_amainpackage.a_o2odata_deal.utils

import a_aa_amainpackage.a_o2odata_deal.config.config._
import com.alibaba.fastjson.JSON
import com.mongodb.spark.MongoSpark
import com.mongodb.spark.config.ReadConfig
import com.mongodb.spark.rdd.MongoRDD
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.bson.Document

/*
*项目名: o2odata_commend
*文件名: from_mongo
*创建者: hancongcong
*创建时间:2019/8/9 16:01
*描述: TODO
*/ object from_mongo {


  /**
    * 开始提数
    *
    * @param sqlContext
    * @param sc
    */
  def spark_mongo_restart(sqlContext: SQLContext, sc: SparkContext): Unit = {
    //原始数据及变量提取
    mongo_detail(sqlContext, sc)
    //补采销量数据提取第一部分
    //mongo_brandapi(sqlContext, sc)
    //补采销量数据提取第二部分
    //mongo_totalsell(sqlContext, sc)
    //补采价格数据提取
    mongo_price_sq(sqlContext, sc)
    for (price_shop_collect <- Array(shop_collection1, shop_collection2, shop_collection3, shop_collection4)) {
      //店铺优惠券数据提取
      mongo_shop_sq(sqlContext, sc, price_shop_collect)
    }
    //新品数据
    mongo_isNewProduct(sqlContext,sc)
    //直播数据
    mongo_livedId(sqlContext,sc)
    mongo_livedDetail(sqlContext,sc)
    //主播集合
    mongo_anchor_id(sqlContext,sc)
    //新增地址
    mongo_newAddress(sqlContext,sc)
  }


  /**
    *
    * 原始数据和变量提取
    * tmall_detail_1907
    *
    * @param sqlContext
    * @param sc
    */
  def mongo_detail(sqlContext: SQLContext, sc: SparkContext): Unit = {

    val detail_vf = ReadConfig(Map(
      "spark.mongodb.input.uri" -> "mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin"
      , "spark.mongodb.input.database" -> "Tmall"
      , "spark.mongodb.input.collection" -> s"${collection_detail}"))

    val deal_field_df = sqlContext.read.json(deal_field(MongoSpark.load(sc, detail_vf)))

    //变量提取
    /*deal_field_df.where("BC_type='B'").select("good_id", "commentCount", "coupon", "crawl_date", "priceText", "preSaleCount", "price_desc", "promotion_price", "sellCount"
      , "categoryId", "brandValueId", "brandName", "shopId", "title", "Base_Info", "detailCategoryName","extendedData")
      .repartition(5).write.json(s"${varpath}")*/
    //原始数据保存
    deal_field_df.repartition(10).write.json(s"${source_obs_data_path}")


  }

  /**
    *
    * 补采销量数据提取第一部分
    * tmall_id_from_brand_api_1907
    *
    * @param sqlContext
    * @param sc
    */
  def mongo_brandapi(sqlContext: SQLContext, sc: SparkContext): Unit = {

    val brandapi_vf = ReadConfig(Map(
      "spark.mongodb.input.uri" -> "mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin"
      , "spark.mongodb.input.database" -> "Tmall"
      , "spark.mongodb.input.collection" -> s"${brand_api_collection}"))
    val brandapi_vf_df = sqlContext.read.json(MongoSpark.load(sc, brandapi_vf).map(line => {
      val nObject = JSON.parseObject(line.toJson())
      nObject.remove("_id")
      nObject.toJSONString
    })).repartition(5).write
      .json(s"${brandpath}")

  }


  /**
    *
    * 补采销量数据提取第二部分
    * tmall_totalsell_day_1907
    *
    * @param sqlContext
    * @param sc
    */
  def mongo_totalsell(sqlContext: SQLContext, sc: SparkContext): Unit = {

    val brandapi_vf = ReadConfig(Map(
      "spark.mongodb.input.uri" -> "mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin"
      , "spark.mongodb.input.database" -> "Tmall"
      , "spark.mongodb.input.collection" -> s"${totalsell_collection}"))
    val brandapi_vf_df = sqlContext.read.json(MongoSpark.load(sc, brandapi_vf).map(line => {
      val nObject = JSON.parseObject(line.toJson())
      nObject.remove("_id")
      nObject.toJSONString
    })).repartition(5).write
      .json(s"${totalsell_path}")

  }

  /**
    *
    * 补采价格数据提取
    * tmall_sq_1906
    *
    * @param sqlContext
    * @param sc
    */
  def mongo_price_sq(sqlContext: SQLContext, sc: SparkContext): Unit = {

    val brandapi_vf = ReadConfig(Map(
      "spark.mongodb.input.uri" -> "mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin"
      , "spark.mongodb.input.database" -> "Tmall"
      , "spark.mongodb.input.collection" -> s"${price_sq_collection}"))
    val brandapi_vf_df = sqlContext.read.json(MongoSpark.load(sc, brandapi_vf).map(line => {
      val nObject = JSON.parseObject(line.toJson())
      nObject.remove("_id")
      nObject.toJSONString
    })).repartition(5).write
      .json(s"${pricesq_path}")



  }

  /**
    *
    * 店铺优惠券数据提取
    * tmall_sq_1906
    *
    * @param sqlContext
    * @param sc
    */
  def mongo_shop_sq(sqlContext: SQLContext, sc: SparkContext, price_shop_collect: String): Unit = {

    val brandapi_vf = ReadConfig(Map(
      "spark.mongodb.input.uri" -> "mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin"
      , "spark.mongodb.input.database" -> "Tmall"
      , "spark.mongodb.input.collection" -> s"${price_shop_collect}"))
    val brandapi_vf_df = sqlContext.read.json(MongoSpark.load(sc, brandapi_vf).map(line => {
      val nObject = JSON.parseObject(line.toJson())
      nObject.remove("_id")
      nObject.toJSONString
    })).repartition(5).write
      .json(clust1 + s"o2o-sourcedata/obs-source-2020/${month}/${platform}/${price_shop_collect}")
  }

  //是否是新品数据
  def mongo_isNewProduct(sqlContext: SQLContext, sc: SparkContext): Unit = {

    val brandapi_vf = ReadConfig(Map(
      "spark.mongodb.input.uri" -> "mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin"
      , "spark.mongodb.input.database" -> "Tmall"
      , "spark.mongodb.input.collection" -> s"${is_newProduct_collection}"))
    val brandapi_vf_df = sqlContext.read.json(MongoSpark.load(sc, brandapi_vf).map(line => {
      val nObject = JSON.parseObject(line.toJson())
      nObject.remove("_id")
      nObject.toJSONString
    })).repartition(5).write
      .json(s"${is_newProduct_path}")
  }

  //直播数据
  def mongo_livedId(sqlContext: SQLContext, sc: SparkContext): Unit = {

    val brandapi_vf = ReadConfig(Map(
      "spark.mongodb.input.uri" -> "mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin"
      , "spark.mongodb.input.database" -> "Tmall"
      , "spark.mongodb.input.collection" -> s"${live_id_collection}"))
    val brandapi_vf_df = sqlContext.read.json(MongoSpark.load(sc, brandapi_vf).map(line => {
      val nObject = JSON.parseObject(line.toJson())
      nObject.remove("_id")
      nObject.toJSONString
    })).repartition(5).write
      .json(s"${tmall_live_id_path}")
  }

  def mongo_livedDetail(sqlContext: SQLContext, sc: SparkContext): Unit = {

    val brandapi_vf = ReadConfig(Map(
      "spark.mongodb.input.uri" -> "mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin"
      , "spark.mongodb.input.database" -> "Tmall"
      , "spark.mongodb.input.collection" -> s"${live_detail_collection}"))
    val brandapi_vf_df = sqlContext.read.json(MongoSpark.load(sc, brandapi_vf).map(line => {
      val nObject = JSON.parseObject(line.toJson())
      nObject.remove("_id")
      nObject.toJSONString
    })).repartition(5).write
      .json(s"${tmall_live_detail_path}")
  }

  def mongo_anchor_id(sqlContext: SQLContext, sc: SparkContext): Unit = {

    val brandapi_vf = ReadConfig(Map(
      "spark.mongodb.input.uri" -> "mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin"
      , "spark.mongodb.input.database" -> "Tmall"
      , "spark.mongodb.input.collection" -> s"tmall_anchor_id"))
    val brandapi_vf_df = sqlContext.read.json(MongoSpark.load(sc, brandapi_vf).map(line => {
      val nObject = JSON.parseObject(line.toJson())
      nObject.remove("_id")
      nObject.toJSONString
    })).repartition(5).write
      .json(s"${tmall_anchor_id_path}")
  }

  //新增地址的拉取
  def mongo_newAddress(sqlContext: SQLContext, sc: SparkContext): Unit = {
    val brandapi_vf = ReadConfig(Map(
      "spark.mongodb.input.uri" -> "mongodb://ob:O2Odata123!@ 192.168.0.149:27017/admin"
      , "spark.mongodb.input.database" -> "Address"
      , "spark.mongodb.input.collection" -> s"tmall_address_200${months}"))
    val brandapi_vf_df = sqlContext.read.json(MongoSpark.load(sc, brandapi_vf).map(line => {
      val nObject = JSON.parseObject(line.toJson())
      nObject.remove("_id")
      nObject.toJSONString
    })).repartition(5).write
      .json(s"${address_new_path}")
  }

  /**
    * 提取天猫的数据
    * 处理　original_cost,starts字段,提取四级name
    * 空调类型 需拆分出来
    */
  def deal_field(rdd: MongoRDD[Document]): RDD[String] = {
    val values = rdd.map(line => {

      val nObject = JSON.parseObject(line.toJson())
      val original_cost = nObject.getOrDefault("original_cost", "-1").toString

      val starts = nObject.getOrDefault("starts", "-1").toString
      var start = ""
      if (!starts.equals("-1")) {
        start = starts.split(" ")(0)
      } else {
        start = "-1"
      }
      var organ = ""
      //      // 切分original_cost字段
      if (!original_cost.equals("-1")) {
        if (original_cost.contains("-")) {
          organ = original_cost.split("-")(0)
        } else {
          organ = original_cost
        }
      } else {
        organ = "-1"
      }
      //每次将baseinfo中吧四级提取出来。情况可能根据不同的品类会有不同的结果。所以需要提取
      var det = "-1"
      val test = nObject.getOrDefault("Base_Info", "-1").toString
      nObject.put("Base_Info", test)
      if (!test.equals("-1")) {
        val base = JSON.parseObject(test)
        val key_arr = base.keySet().toArray()
        for (i <- key_arr) {
          val key = i.toString
          if (key.equals("分类")) {
            det = base.get(key).toString
            nObject.put("detailCategoryName", det)
          } else if (key.equals("类型")) {
            det = base.get(key).toString
            nObject.put("detailCategoryName", det)
          } else if (key.equals("品类")) {
            det = base.get(key).toString
            nObject.put("detailCategoryName", det)
          } else if (key.equals("种类")) {
            det = base.get(key).toString
            nObject.put("detailCategoryName", det)
          } else if (key.contains("分类")) {
            det = base.get(key).toString
            nObject.put("detailCategoryName", det)
          } else if (key.contains("类型")) {
            det = base.get(key).toString
            nObject.put("detailCategoryName", det)
          } else {
            nObject.put("detailCategoryName", det)
          }
        }
      } else {
        nObject.put("detailCategoryName", det)
      }
      nObject.remove("_id")
      nObject.put("original_cost", organ)
      nObject.put("starts", start)
      nObject.toString()
    })
    values
  }


}
